/*
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements.  See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package consumer

import (
	"go.uber.org/atomic"
	"strings"
	"time"

	"gitee.com/zackeus/go-boot/rocketmq/hooks"
	"gitee.com/zackeus/go-boot/rocketmq/internal"
	"gitee.com/zackeus/go-boot/rocketmq/primitive"
)

type consumerOptions struct {
	internal.ClientOptions

	/**
	 * Backtracking consumption time with second precision. Time format is
	 * 20131223171201<br>
	 * Implying Seventeen twelve and 01 seconds on December 23, 2013 year<br>
	 * Default backtracking consumption time Half an hour ago.
	 */
	ConsumeTimestamp string

	// The socket timeout in milliseconds
	ConsumerPullTimeout time.Duration

	// Concurrently max span offset.it has no effect on sequential consumption
	ConsumeConcurrentlyMaxSpan int

	// Flow control threshold on queue level, each message queue will cache at most 1000 messages by default,
	// Consider the {PullBatchSize}, the instantaneous value may exceed the limit
	PullThresholdForQueue int64

	// Limit the cached message size on queue level, each message queue will cache at most 100 MiB messages by default,
	// Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
	//
	// The size of a message only measured by message body, so it's not accurate
	PullThresholdSizeForQueue int

	// Flow control threshold on topic level, default value is -1(Unlimited)
	//
	// The value of {@code pullThresholdForQueue} will be overwritten and calculated based on
	// {@code pullThresholdForTopic} if it isn't unlimited
	//
	// For example, if the value of pullThresholdForTopic is 1000 and 10 message queues are assigned to this consumer,
	// then pullThresholdForQueue will be set to 100
	PullThresholdForTopic int

	// Limit the cached message size on topic level, default value is -1 MiB(Unlimited)
	//
	// The value of {@code pullThresholdSizeForQueue} will be overwritten and calculated based on
	// {@code pullThresholdSizeForTopic} if it isn't unlimited
	//
	// For example, if the value of pullThresholdSizeForTopic is 1000 MiB and 10 message queues are
	// assigned to this consumer, then pullThresholdSizeForQueue will be set to 100 MiB
	PullThresholdSizeForTopic int

	// Message pull Interval
	PullInterval atomic.Duration

	// Message consumer tps
	ConsumeTPS atomic.Int32

	// Batch consumption size
	ConsumeMessageBatchMaxSize int

	// Batch pull size
	PullBatchSize atomic.Int32

	// Whether update subscription relationship when every pull
	PostSubscriptionWhenPull bool

	// Max re-consume times. -1 means 16 times.
	//
	// If messages are re-consumed more than {@link #maxReconsumeTimes} before Success, it's be directed to a deletion
	// queue waiting.
	MaxReconsumeTimes int32

	// Suspending pulling time for cases requiring slow pulling like flow-control scenario.
	SuspendCurrentQueueTimeMillis time.Duration

	// Maximum amount of time a message may block the consuming thread.
	ConsumeTimeout time.Duration

	ConsumerModel  MessageModel
	Strategy       AllocateStrategy
	ConsumeOrderly bool
	FromWhere      ConsumeFromWhere

	Interceptors []primitive.Interceptor
	// TODO traceDispatcher
	MaxTimeConsumeContinuously time.Duration
	//
	AutoCommit            bool
	RebalanceLockInterval time.Duration

	Resolver primitive.NsResolver

	ConsumeGoroutineNums int

	filterMessageHooks []hooks.FilterMessageHook

	Limiter Limiter

	TraceDispatcher internal.TraceDispatcher
}

func defaultPushConsumerOptions() consumerOptions {
	opts := consumerOptions{
		ClientOptions:              internal.DefaultClientOptions(),
		Strategy:                   AllocateByAveragely,
		MaxTimeConsumeContinuously: time.Duration(60 * time.Second),
		RebalanceLockInterval:      20 * time.Second,
		MaxReconsumeTimes:          -1,
		ConsumerModel:              Clustering,
		AutoCommit:                 true,
		Resolver:                   primitive.NewHttpResolver("DEFAULT"),
		ConsumeTimestamp:           time.Now().Add(time.Minute * (-30)).Format("20060102150405"),
		ConsumeTimeout:             15 * time.Minute,
		ConsumeGoroutineNums:       20,
	}
	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
	return opts
}

type Option func(*consumerOptions)

func defaultPullConsumerOptions() consumerOptions {
	opts := consumerOptions{
		ClientOptions: internal.DefaultClientOptions(),
		Resolver:      primitive.NewHttpResolver("DEFAULT"),
		ConsumerModel: Clustering,
		Strategy:      AllocateByAveragely,
	}
	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
	return opts
}

// WithConsumerModel 消费模式
func WithConsumerModel(m MessageModel) Option {
	return func(options *consumerOptions) {
		options.ConsumerModel = m
	}
}

// WithConsumeFromWhere 消费起始位置
func WithConsumeFromWhere(w ConsumeFromWhere) Option {
	return func(options *consumerOptions) {
		options.FromWhere = w
	}
}

func WithConsumerOrder(order bool) Option {
	return func(options *consumerOptions) {
		options.ConsumeOrderly = order
	}
}

// WithConsumeMessageBatchMaxSize 单次消费批次大小
func WithConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize int) Option {
	return func(options *consumerOptions) {
		options.ConsumeMessageBatchMaxSize = consumeMessageBatchMaxSize
	}
}

// WithConsumeTimestamp ConsumeFromTimestamp 时使用 指定消费时间点(格式 yyyyMMddhhmmss)
// 如果从消息进度服务OffsetStore读取到MessageQueue中的偏移量大于等于0，则使用读取到的偏移量，只有读取到的偏移量小于0时上述策略才会生效
func WithConsumeTimestamp(consumeTimestamp string) Option {
	return func(options *consumerOptions) {
		options.ConsumeTimestamp = consumeTimestamp
	}
}

func WithConsumerPullTimeout(consumerPullTimeout time.Duration) Option {
	return func(options *consumerOptions) {
		options.ConsumerPullTimeout = consumerPullTimeout
	}
}

// WithConsumeConcurrentlyMaxSpan 并发消费允许的最大跨度(顺序消费无影响)
func WithConsumeConcurrentlyMaxSpan(consumeConcurrentlyMaxSpan int) Option {
	return func(options *consumerOptions) {
		options.ConsumeConcurrentlyMaxSpan = consumeConcurrentlyMaxSpan
	}
}

func WithPullThresholdForQueue(pullThresholdForQueue int64) Option {
	return func(options *consumerOptions) {
		options.PullThresholdForQueue = pullThresholdForQueue
	}
}

func WithPullThresholdSizeForQueue(pullThresholdSizeForQueue int) Option {
	return func(options *consumerOptions) {
		options.PullThresholdSizeForQueue = pullThresholdSizeForQueue
	}
}

func WithPullThresholdForTopic(pullThresholdForTopic int) Option {
	return func(options *consumerOptions) {
		options.PullThresholdForTopic = pullThresholdForTopic
	}
}

func WithPullThresholdSizeForTopic(pullThresholdSizeForTopic int) Option {
	return func(options *consumerOptions) {
		options.PullThresholdSizeForTopic = pullThresholdSizeForTopic
	}
}

// WithInterceptor returns a ConsumerOption that specifies the chained interceptor for consumer.
// The first interceptor will be the outer most, while the last interceptor will be the inner most wrapper
// around the real call.
func WithInterceptor(fs ...primitive.Interceptor) Option {
	return func(options *consumerOptions) {
		options.Interceptors = append(options.Interceptors, fs...)
	}
}

// WithGroupName set group name address
func WithGroupName(group string) Option {
	return func(opts *consumerOptions) {
		if group == "" {
			return
		}
		opts.GroupName = group
	}
}

func WithInstance(name string) Option {
	return func(options *consumerOptions) {
		options.InstanceName = name
	}
}

// WithNamespace set the namespace of consumer
func WithNamespace(namespace string) Option {
	return func(opts *consumerOptions) {
		opts.Namespace = namespace
	}
}

func WithVIPChannel(enable bool) Option {
	return func(opts *consumerOptions) {
		opts.VIPChannelEnabled = enable
	}
}

// WithRetry return a Option that specifies the retry times when send failed.
// TODO: use retry middleware instead
func WithRetry(retries int) Option {
	return func(opts *consumerOptions) {
		opts.RetryTimes = retries
	}
}

func WithCredentials(c primitive.Credentials) Option {
	return func(options *consumerOptions) {
		options.ClientOptions.Credentials = c
	}
}

// WithMaxReconsumeTimes 消费失败最大重试次数(只针对集群消费模式生效 并发消费间隔时间根据重试次数阶梯变化)
// 达到最大次数，消息将会发送至死信队列，不再重试
func WithMaxReconsumeTimes(times int32) Option {
	return func(opts *consumerOptions) {
		opts.MaxReconsumeTimes = times
	}
}

// WithStrategy 消息消费分配策略
func WithStrategy(strategy AllocateStrategy) Option {
	return func(opts *consumerOptions) {
		opts.Strategy = strategy
	}
}

// WithPullBatchSize 每次 pull 消息大小
func WithPullBatchSize(batchSize int32) Option {
	return func(options *consumerOptions) {
		options.PullBatchSize.Store(batchSize)
	}
}

func WithRebalanceLockInterval(interval time.Duration) Option {
	return func(options *consumerOptions) {
		options.RebalanceLockInterval = interval
	}
}

func WithAutoCommit(auto bool) Option {
	return func(options *consumerOptions) {
		options.AutoCommit = auto
	}
}

// WithSuspendCurrentQueueTimeMillis 消费失败重试再次被投递给Consumer消费的间隔时间(只在顺序消费中起作用)
func WithSuspendCurrentQueueTimeMillis(suspendT time.Duration) Option {
	return func(options *consumerOptions) {
		options.SuspendCurrentQueueTimeMillis = suspendT
	}
}

// WithPullInterval 检查拉取消息的间隔时间
func WithPullInterval(interval time.Duration) Option {
	return func(options *consumerOptions) {
		options.PullInterval.Store(interval)
	}
}

// WithConsumeTPS 单机消费 TPS(会根据此值动态调整 PullInterval 实现拉取速率控制)
// Principle of flow control: pull TPS = 1000ms/PullInterval * BatchSize * len(allocateResult)
func WithConsumeTPS(tps int32) Option {
	return func(options *consumerOptions) {
		options.ConsumeTPS.Store(tps)
	}
}

// WithNsResolver set nameserver resolver to fetch nameserver addr
func WithNsResolver(resolver primitive.NsResolver) Option {
	return func(options *consumerOptions) {
		options.Resolver = resolver
	}
}

// WithNameServer set NameServer address, only support one NameServer cluster in alpha2
func WithNameServer(nameServers primitive.NamesrvAddr) Option {
	return func(options *consumerOptions) {
		options.Resolver = primitive.NewPassthroughResolver(nameServers)
	}
}

// WithNameServerDomain set NameServer domain
func WithNameServerDomain(nameServerUrl string) Option {
	return func(opts *consumerOptions) {
		h := primitive.NewHttpResolver("DEFAULT", nameServerUrl)
		if opts.UnitName != "" {
			h.DomainWithUnit(opts.UnitName)
		}
		opts.Resolver = h
	}
}

// WithUnitName set the name of specified unit
func WithUnitName(unitName string) Option {
	return func(opts *consumerOptions) {
		opts.UnitName = strings.TrimSpace(unitName)
		if ns, ok := opts.Resolver.(*primitive.HttpResolver); ok {
			ns.DomainWithUnit(opts.UnitName)
		}
	}
}

// WithConsumeTimeout 消费超时时间(集群无序模式有效)
func WithConsumeTimeout(timeout time.Duration) Option {
	return func(opts *consumerOptions) {
		opts.ConsumeTimeout = timeout
	}
}

// WithConsumeGoroutineNums 消费线程数 消费速度取决于 ConsumeGoroutineNums * ConsumeMessageBatchMaxSize
func WithConsumeGoroutineNums(nums int) Option {
	return func(opts *consumerOptions) {
		opts.ConsumeGoroutineNums = nums
	}
}

func WithFilterMessageHook(hooks []hooks.FilterMessageHook) Option {
	return func(opts *consumerOptions) {
		opts.filterMessageHooks = hooks
	}
}

func WithLimiter(limiter Limiter) Option {
	return func(opts *consumerOptions) {
		opts.Limiter = limiter
	}
}
